package com.lxl.testHd.myoutputformat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * @author ：e_lixilin
 * @date ：2022/2/24 9:50
 * @description：
 * @modified By：
 */
public class SelfOutputFormat extends FileOutputFormat<Text, NullWritable> {
    @Override
    public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
        Configuration conf = job.getConfiguration();
        FileSystem fs = FileSystem.get(conf);
        FSDataOutputStream smallCarOut = fs.create(new Path("D:\\big-data\\etl\\myoutputformat\\out\\small.txt"));
        FSDataOutputStream normalCarOut = fs.create(new Path("D:\\big-data\\etl\\myoutputformat\\out\\normalCarOut.txt"));
        FSDataOutputStream bigCarOut = fs.create(new Path("D:\\big-data\\etl\\myoutputformat\\out\\bigCarOut.txt"));
        FSDataOutputStream superCarOut = fs.create(new Path("D:\\big-data\\etl\\myoutputformat\\out\\superCarOut.txt"));
        return new MyRecordWriter(smallCarOut,normalCarOut,bigCarOut,superCarOut);
    }

    static class MyRecordWriter extends RecordWriter<Text, NullWritable> {
        FSDataOutputStream smallCarOut = null;
        FSDataOutputStream normalCarOut = null;
        FSDataOutputStream bigCarOut = null;
        FSDataOutputStream superCarOut = null;

        public MyRecordWriter(FSDataOutputStream smallCarOut, FSDataOutputStream normalCarOut,
                              FSDataOutputStream bigCarOut, FSDataOutputStream superCarOut) {
            this.smallCarOut = smallCarOut;
            this.normalCarOut = normalCarOut;
            this.bigCarOut = bigCarOut;
            this.superCarOut = superCarOut;
        }

        @Override
        public void write(Text key, NullWritable value) throws IOException, InterruptedException {
            // 不同的车型信息写入到不同的文件里, key 为mapper写入的一行内容:
            // 2018-09-15.06:31:31	2018-09-15.08:58:58	262	中型	浙KL69T0
            String line = key.toString();
            String carType = line.split("\t")[3];
            switch (carType) {
                case "小型":
                    smallCarOut.writeChars(line);
                    smallCarOut.writeChars("\r\n");
                    break;
                case "中型":
                    normalCarOut.writeChars(line);
                    normalCarOut.writeChars("\r\n");
                    break;
                case "大型":
                    bigCarOut.write(line.getBytes());
                    bigCarOut.writeChars("\r\n");
                    break;
                case "特大型":
                    superCarOut.write(line.getBytes());
                    superCarOut.writeChars("\r\n");
                    break;
                default:
                    System.out.println("车型错误");
                    break;
            }
        }

        @Override
        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
            IOUtils.closeStream(smallCarOut);
            IOUtils.closeStream(normalCarOut);
            IOUtils.closeStream(bigCarOut);
            IOUtils.closeStream(superCarOut);
        }
    }
}
